/*===================================================================================
 Copyright (c) 2004-2007 by www.koubei.com, All rights reserved.
 8f., HuaXing technology building, 477# wensan road, HangZhou, China
 
 This software is the confidential and proprietary information of 
 Koubei.com, Inc. ("Confidential Information"). You shall not disclose 
 such Confidential Information and shall use it only in accordance 
 with the terms of the license agreement you entered into with Koubei.com, Inc.
===================================================================================*/
package com.koubei.tools.priceindex.framework;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobConfigurable;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

/**
 * @Author Peltason
 * @Date 2008-6-6
 * @Description
 * @Others
 * @FunctionList ...
 * @History
 */
public class MultiLineInputFormat extends FileInputFormat<LongWritable, Text> implements JobConfigurable {

	private CompressionCodecFactory compressionCodecs = null;
	  
	public void configure(JobConf conf) {
		compressionCodecs = new CompressionCodecFactory(conf);
	}
	  
	protected boolean isSplitable(FileSystem fs, Path file) {
	    return compressionCodecs.getCodec(file) == null;
	}

	public RecordReader<LongWritable, Text> getRecordReader(InputSplit genericSplit, JobConf job,
			Reporter reporter) throws IOException {	    
	    reporter.setStatus(genericSplit.toString());
	    return new MultiLineRecordReader(job, (FileSplit) genericSplit);
	}
	
}
